with must_center_step1 as (
 select
     t.start_code  as start_code
    ,t1.name  as end_name 
    ,tt.end_name as end_name1
    ,t2.network_type as end_name1_network_type
    ,nvl(tt.delivery_networklist,'99999')  as delivery_networklist
    ,sum( case when  t1.network_type = 4 then 1 else 0 end )over( partition by t.start_code) as network_flg
    ,t1.network_type
from
    jms_dim.dim_yl_tms_branch_line_base_dt t
   left join jms_dim.dim_tms_branch_delivery_network_base_dt tt on t.id = tt.line_id
   left join jms_dim.dim_network_whole_massage t1 on t.end_code = t1.code
   left join jms_dim.dim_network_whole_massage t2 on t2.code = tt.end_code

where
     t.is_enable=1 and t.dt =  '{{ execution_date | cst_ds }}'
    and tt.dt =  '{{ execution_date | cst_ds }}'
),must_center_step2 as (
    select 
        t1.start_code,
        str_to_map(concat_ws(',', collect_set(t1.tmp_map))) as all_map,
        max(network_flg) as network_flg,
        max(null_end_name) null_end_name,
        max(null_end_name1) null_end_name1
    from (
            select 
              t.start_code
             ,all_end
             ,concat_ws(':', all_end,concat_ws('-', collect_set( case when t.network_type =4 then t.end_name end ))) tmp_map
             ,concat_ws('-', collect_set( case when t.delivery_networklist ='99999' and t.network_type =4 then t.end_name end )) as null_end_name
             ,concat_ws('-', collect_set( case when t.end_name1_network_type =4 then t.end_name1 end )) as null_end_name1

             ,max( network_flg )  network_flg
            from must_center_step1 t
            lateral view explode(split(t.delivery_networklist,',')) tb_id as all_end
           group by t.start_code,all_end
        ) t1
group by   t1.start_code
)

,
first_code_date as (
    select code,max(center_code) center_code
    from jms_dim.dim_lmdm_sys_first_code
    where is_enable=1 and is_delete = 1
    group by code
)
,dm_site_wrong_detail_dt_tmp as (
select /*+ broadcast(t5),broadcast(t6)*/
 to_date(t2.taking_early_scantime)  as date_time
,t2.waybill_no
,t2.taking_early_scantime as taking_scantime
,t2.taking_early_scansitecode as  taking_site_code

,tt.name  as taking_site_name
,tt.agent_code  
,tt.agent_name 
,tt.fran_code 
,tt.fran_name 
,t2.first_center_network_code
,tt2.name  as first_center_network_name
,tt4.center_code as end_center_network_code
,tt3.name as end_center_network_name
, concat_ws('-',if( t5.null_end_name ='' , null , t5.null_end_name ) ,if(t5.all_map[tt4.center_code]='',t5.null_end_name1,t5.all_map[tt4.center_code])
               ,if( t6.null_end_name ='' , null , t6.null_end_name ) ,if(t6.all_map[tt4.center_code]='',t6.null_end_name1,t6.all_map[tt4.center_code])) as  must_center_name
,to_date(t2.taking_early_scantime)  as dt
from jms_dwd.dwd_wide_unsign_summary_waybill_dt t2 
left join must_center_step2 t5 on  t2.taking_early_scansitecode=t5.start_code       --网点
left join must_center_step2 t6 on  if (greatest(t2.first_nodal_arrival_early_time,t2.first_nodal_send_later_time) < least(t2.first_center_arrival_early_time,t2.first_center_send_later_time),t2.first_nodal_network_code,t2.waybill_no)=t6.start_code       --集散
left join first_code_date tt4 on tt4.code=t2.first_code
left join jms_dim.dim_network_whole_massage tt on tt.code=t2.taking_early_scansitecode
left join jms_dim.dim_network_whole_massage tt2 on tt2.code=t2.first_center_network_code
left join jms_dim.dim_network_whole_massage tt3 on tt3.code=tt4.center_code

where t2.dt>='{{ execution_date | date_add(-20)  | cst_ds}}'
     and to_date(t2.taking_early_scantime)>='{{ execution_date | date_add(-14) | cst_ds }}'
     and to_date(t2.taking_early_scantime)<='{{ execution_date | cst_ds }}'
     and t2.first_center_network_code is not null
     and t2.taking_early_scantime is not null
     and t2.is_reback =0
)
insert overwrite table jms_dm.dm_site_wrong_detail_dt partition ( dt )
select 
 t.dt as date_time
,t.waybill_no
,max(t.taking_scantime) as taking_scantime
,max(t.taking_site_code) as  taking_site_code
,max(t.taking_site_name)  as taking_site_name
,max(t.agent_code) as agent_code 
,max(t.agent_name) as agent_name
,max(t.fran_code) as fran_code
,max(t.fran_name) as fran_name
,max(t.first_center_network_code) as first_center_network_code
,max(t.first_center_network_name)  as first_center_network_name
,max(t.end_center_network_code) as end_center_network_code
,max(t.end_center_network_name) as end_center_network_name
,concat_ws('-', collect_set(must_center_name1))   as must_center_name
,max(case when  !array_contains(split(t.must_center_name,'-'),t.first_center_network_name) and t.must_center_name is not null and t.must_center_name  <> ''  then 1 else 0 end ) as is_wrong
,t.dt
from dm_site_wrong_detail_dt_tmp t
lateral view explode (split(t.must_center_name,'-')) as must_center_name1
group by t.waybill_no,t.dt
distribute by dt, abs(hash(waybill_no)) % 100 
;
 